package com.fc.test.compare;

import com.fc.test.model.auto.TableConfig;
import com.fc.test.model.custom.EtlHistablePartitionsConfiguration;
import com.fc.test.util.StringUtils;
import com.google.common.base.Predicates;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Predicate;
import javax.sql.DataSource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created by Liaopan on 2019/7/4.
 */
public class Launcher {

    private static final Logger LOG = LoggerFactory.getLogger(Launcher.class);
    /**
     * 任务运行的并行个数
     */
    private ExecutorService taskPools = null;

    /**
     * 任务table
     */
    private List<TableConfig> tables;

    private DataSource impalaDataSource;

    private List<EtlHistablePartitionsConfiguration> epConfig;

    public void initialize(int poolSize, List<TableConfig> tables) {
        taskPools = Executors.newFixedThreadPool(poolSize);
        this.tables = tables;
    }

    public void setImpalaDataSource(DataSource dataSource) {
        this.impalaDataSource = dataSource;
    }

    public void setHisTablePartitionConfig(List<EtlHistablePartitionsConfiguration> list) {
        this.epConfig = list;
    }

    public void doTheTask() {
        LOG.debug("共加载表个数：" + tables.size() + ",开始时间：" + LocalDateTime.now());
        for (TableConfig table : tables) {
            String sourceTable = table.getSourceTable();
            String tableName = "";
            String schema = "";
            if (sourceTable.contains(".")) {
                tableName = StringUtils.substringAfter(sourceTable, ".");
                schema = StringUtils.substringBefore(sourceTable, ".");
            }
            final String schemaFinal = schema;
            final String tableNameFinal = tableName;
            String timeColumn = epConfig.stream().filter( ep -> ep.getHisDbName().equalsIgnoreCase(schemaFinal)
                    && ep.getHisTbName().equalsIgnoreCase(tableNameFinal)).findFirst()
                    .map(EtlHistablePartitionsConfiguration::getHisTbPartitionColumnName).orElse("");
            table.setPartitionColumn(timeColumn);
            // 增加全量/增量/区间的判断
            /**
             * 默认有时间列的情况
             * 增量： 开始时间是 当前时间-(range+1), 结束时间 =  （当前时间-1） （单位：天）
             * 全量： 开始时间 = 1900-01-01 (默认写死) 结束时间 = （当前时间-1） （单位：天）
             * 区间： 根据设置的时间来
             */
            LocalDate localDate = LocalDate.now();
            if(JobConstants.RunType.ALL.getCode().toString().equalsIgnoreCase(table.getRunType())) {
                localDate = LocalDate.of(1900,1,1);
                table.setStartDate(null);
                table.setEndDate(null);
            }else if (JobConstants.RunType.INCR.getCode().toString().equalsIgnoreCase(table.getRunType())) {

                table.setStartDate(Date.from(localDate.minusDays(table.getRange() + 1)
                        .atStartOfDay().atZone(ZoneId.systemDefault()).toInstant()));
                table.setEndDate(Date.from(localDate.minusDays(1).atStartOfDay().atZone(ZoneId.systemDefault()).toInstant()));
            }

            taskPools.submit(new CompareTask(table,new CompareColumnsReturnMap()));
        }
        taskPools.shutdown();
    }
}
